Apache Flink 如何计算实时热门商品 | 每天，遇到更好的你

目录

Apache Flink 实验: 如何计算实时热门商品

本实验目标

如何基于 EventTime 处理，如何指定 Watermark
如何使用 Flink 灵活的 Window API
何时需要用到 State，以及如何使用
如何使用 ProcessFunction 实现 TopN 功能

实验案例介绍

本案例将实现一个”实时热门商品”的需求，我们可以将”实时热门商品”翻译成程序员更好理解的需求:

每隔 5 分钟输出最近一小时内点击量最多的前 N 个商品。将这个需求进行分解我们大概要做这么几件事情:

抽取出业务时间戳，告诉 Flink 框架基于业务时间做窗口
过滤出点击行为数据
按一小时的窗口大小，每 5 分钟统计一次，做滑动窗口聚合(Sliding Window)
按每个窗口聚合，输出每个窗口中点击量前 N 名的商品

数据准备

这里我们准备了一份淘宝用户行为数据集(来自阿里云天池公开数据集，特别感谢)。本数据集包含了淘宝上某一天随机一百万用户的所有行为(包括点击、购买、加购、收藏)。数据集的组织形式和 MovieLens-20M 类似，即数据集的每一行表示一条用户行为，由用户 ID、商品 ID、商品类目 ID、行为类型和时间戳组成，并以逗号分隔。关于数据集中每一列的详细描述如下:

列名称	说明
用户ID	整数类型，加密后的用户ID
商品ID	整数类型，加密后的商品ID
商品类目ID	整数类型，加密后的商品所属类目ID
行为类型	字符串，枚举类型，包括(‘pv’,’buy’,’cart’,’fav’)
时间戳	行为发生的时间戳，单位秒

你可以通过下面的命令下载数据集到项目的 resources 目录下:

1
2
3

bash
sparsematrix:flink-demo matrix$ cd src/main/resources/
wget https://raw.githubusercontent.com/wuchong/my-flink-project/master/src/main/resources/UserBehavior.csv > UserBehavior.csv

这里是否使用 curl 命令下载数据并不重要，你也可以使用 wget 命令或者直接访问链接下载数据。关键是，将数据文件保存到项目的 resources 目录下，方便应用程序访问。

编写程序

在 src/main/java/com.matrix.flink 下创建 HotItems.java 文件:

java
package com.matrix.flink;

/**
 * 实时热门商品
 */
public class HotItems {

    public static void main(String[] args) throws Exception {

    }
}

与上文一样，我们会一步步往里面填充代码。第一步仍然是创建一个 StreamExecutionEnvironment，我们把它添加到 main 函数中。

1 2	java StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

创建模拟数据源

在数据准备章节，我们已经将测试的数据集下载到本地了。由于是一个 csv 文件，我们将使用 CsvInputFormat 创建模拟数据源。

注:虽然一个流式应用应该是一个一直运行着的程序，需要消费一个无限数据源。但是在本案例教程中，为了省去构建真实数据源的繁琐，我们使用了文件来模拟真实数据源，这并不影响下文要介绍的知识点。这也是一种本地验证 Flink 应用程序正确性的常用方式。

我们先创建一个 UserBehavior 的 POJO 类(所有成员变量声明成 public 便是 POJO 类)，强类型化后能方便后续的处理。

java
package com.matrix.flink.model;

/**
 * 用户行为实体类
 */
public class UserBehavior {

    // 用户ID
    public long userId;
    // 商品ID
    public long itemID;
    // 用户行为，包括("pv","buy")
    public String behavior;
    // 行为发生的时间戳，单位秒
    public long timestamp;

    public long getUserId() {
        return userId;
    }

    public long getItemID() {
        return itemID;
    }

    public String getBehavior() {
        return behavior;
    }

    public long getTimestamp() {
        return timestamp;
    }

    public void setUserId(long userId) {
        this.userId = userId;
    }

    public void setItemID(long itemID) {
        this.itemID = itemID;
    }

    public void setBehavior(String behavior) {
        this.behavior = behavior;
    }

    public void setTimestamp(long timestamp) {
        this.timestamp = timestamp;
    }
}

接下来我们就可以创建一个PojoCsvInputFormat了，这是一个读取 csv 文件并将每一行转成指定 POJO 类型(在我们案例中是 UserBehavior)的输入器。

java
// UserBehavior.csv 的本地文件路径
URL fileUrl = HotItems2.class.getClassLoader().getResource("UserBehavior.csv"); Path filePath = Path.fromLocalFile(new File(fileUrl.toURI()));
// 抽取 UserBehavior 的 TypeInformation，是一个 PojoTypeInfo PojoTypeInfo<UserBehavior> pojoType = (PojoTypeInfo<UserBehavior>) TypeExtractor.createTypeInfo(UserBehavior.class);
// 由于 Java 反射抽取出的字段顺序是不确定的，需要显式指定下文件中字段的顺序
String[] fieldOrder = new String[]{"userId", "itemId", "categoryId", "behavior", "timestamp"};
// 创建 PojoCsvInputFormat
PojoCsvInputFormat<UserBehavior> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);